package com.nepu.gmall.realtime.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONAware;
import com.alibaba.fastjson.JSONObject;
import com.nepu.gmall.realtime.util.KafkaUtils;
import com.sun.xml.internal.bind.v2.TODO;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.PatternTimeoutFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.IterativeCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.List;
import java.util.Map;

/**
 * 本类处理的是流量域用户跳出事物事实表的数据。
 * 本类对数据的处理逻辑如下：
 *  （1）用户跳出指的是当前用户只访问了一个页面就直接退出应用了。
 *  （2）在离线数仓中，我们对于该指标的计算主要是通过session_id找到一个会话中只有一个页面的访问记录。
 *      session_id的构建是，mid_firstTs,firstTs:指的是本次会话的开始时间
 *  （3）在实时数仓中，我们不确定之后还会不会有页面的访问，所以无法构架sessionID,这里采用的解决办法是。
 *      3.1->采用的是CEP来解决来做复杂事件的处理,CEP的本质是（状态编程+within开窗）。所以我们将以如下的定义。
 *          如果两次连续的访问的数据他的last_page_id == null ，那么第一条数据作为一个单独的会话
 *          如果有一个事件已经满足就，但是另一个事件是迟迟没有到来，那么这里设置一个窗口的超时的时间是10s,如果10s下一个事件还没有到来，就认为这次访问是一个独立的会话
 *          将超时的事件写入到侧输出流中
 *  （4）将主流和侧输出流中的数据合并写入到kafka对应的分区中。
 *
 *  测试时数据的流向
 *  web/app --> 日志服务器 --> kafka --> BaseLogApp --> kafka --> DwdTrafficUserJumpDetail --> kafka
 *  mock --> f1.sh --> kafka(zookeeper) --> BaseLogApp --> kafka --> DwdTrafficUserJumpDetail --> kafka
 * @author chenshuaijun
 * @create 2023-02-24 16:49
 */
public class DwdTrafficUserJumpDetail {

    public static void main(String[] args) throws Exception {
        // TODO 1、加载环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 生产环境下是一定不会将任务的并行度设置为1的，这里具体的设置是和我们等下要读取的kafka的相应的主题的分区的个数相同
        env.setParallelism(1);
        // 设置checkpoint的信息：设置checkpoint的间隔是5分钟,并且checkpoint的级别是精确一次性
        /*env.enableCheckpointing(5 * 60000L, CheckpointingMode.EXACTLY_ONCE);
        // 设置checkpoint的超时时间是10分钟
        env.getCheckpointConfig().setCheckpointTimeout(10 * 60000L);
        // 设置外部检查点。可以将检查点的元数据信息定期写入外部系统，这样当job失败时，检查点不会被清除。这样如果job失败，可以从检查点恢复job。
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // 设置checkpoint的重启的策略
        env.setRestartStrategy(RestartStrategies.failureRateRestart(10, Time.of(1L, TimeUnit.DAYS), Time.of(3L, TimeUnit.MINUTES)));
        // 设置两个checkpoint之间的最小的间隔时间
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
        // 设置状态后端: 设置状态后端为内存级别
        env.setStateBackend(new HashMapStateBackend());
        // 设置checkpoint的存储的路径
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/checkpoint");
        // 因为我们的HDFS只有atguigu用户才能够操作，所以要将用户设置为atguigu
        System.setProperty("HADOOP_USER_NAME", "atguigu");*/

        // TODO 2、从kafka中消费数据
        String topic  = "dwd_traffic_page_log";
        String groupId = "DwdTrafficUserJumpDetail";
        DataStreamSource<String> kafkaDataStream = env.addSource(KafkaUtils.getKafkaConsumer(topic, groupId));
        // TODO 3、转换数据结构
        SingleOutputStreamOperator<JSONObject> jsonObjStream = kafkaDataStream.map(JSON::parseObject);
        // TODO 3.1 提起watermark，讲数据进行分组
        KeyedStream<JSONObject, String> jsonObjectStringKeyedStream = jsonObjStream.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
            @Override
            public long extractTimestamp(JSONObject element, long recordTimestamp) {
                return element.getLong("ts");
            }
        })).keyBy(jsonData -> jsonData.getJSONObject("common").getString("mid"));
        // TODO 4、定义模式
        Pattern<JSONObject, JSONObject> pattern = Pattern.<JSONObject>begin("start")
                .where(new IterativeCondition<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject value, Context<JSONObject> ctx) throws Exception {
                        return value.getJSONObject("page").getString("last_page_id") == null;
                    }
                }).next("end")
                .where(new IterativeCondition<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject value, Context<JSONObject> ctx) throws Exception {
                        return value.getJSONObject("page").getString("last_page_id") == null;
                    }
                }).within(Time.seconds(10));
        // 简洁版：上面我们定义的两个where的规则是相同的，我们可以使用循环模式来简化我们的代码
        /*Pattern<JSONObject, JSONObject> pattern = Pattern.<JSONObject>begin("start")
                .where(new IterativeCondition<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject value, Context<JSONObject> ctx) throws Exception {
                        return value.getJSONObject("page").getString("last_page_id") == null;
                    }
                })
                .times(2)
                .consecutive() // 定义为严格近邻关系，要不然默认使用的是宽松近邻
                .within(Time.seconds(10));*/
        // TODO 5、匹配数据
        PatternStream<JSONObject> patternStream = CEP.pattern(jsonObjectStringKeyedStream, pattern);
        // TODO 6、获取数据
        OutputTag<JSONObject> outputTag = new OutputTag<JSONObject>("timeoutData") {};
        SingleOutputStreamOperator<JSONObject> selectDataStream = patternStream.select(outputTag, new PatternTimeoutFunction<JSONObject, JSONObject>() {
            @Override
            public JSONObject timeout(Map<String, List<JSONObject>> pattern, long timeoutTimestamp) throws Exception {
                return pattern.get("start").get(0);
            }
        }, new PatternSelectFunction<JSONObject, JSONObject>() {
            @Override
            public JSONObject select(Map<String, List<JSONObject>> pattern) throws Exception {
                JSONObject selectData = pattern.get("start").get(0);
                return selectData;
            }
        });
        // TODO 7、合并数据
        DataStream<JSONObject> timeoutStream = selectDataStream.getSideOutput(outputTag);

        DataStream<JSONObject> unionDtaStream = selectDataStream.union(timeoutStream);
        // TODO 8、将数据写入到kafka
        timeoutStream.print("超时数据--->");
        selectDataStream.print("正常数据--->");
        String targetTopic = "dwd_traffic_user_jump_detail";
        unionDtaStream.map(JSONAware::toJSONString).addSink(KafkaUtils.getFlinkKafkaProducer(targetTopic));
        // TODO 9、运行程序
        env.execute();
    }
}
